/*
 * alivecheck.c - alive check
 * Clunix, KyungWoon Cho, cezanne@clunix.com, 2006.2
 *
 * Copyright (c) 2001-2008 by Clunix Inc. All Rights Reserved.
 */

#include "config.h"

#include "ebcd.h"

ecm_uint32	ebcd_alivecheck_timeout = EBCD_DEFAULT_ALIVECHECK_TIMEOUT;

static ecm_mutex_t	alivecheck_mutex;
static ecm_cond_t	alivecheck_cond;

static LIST_HEAD(sent_pchs);
static LIST_HEAD(recv_pchs);

static ecm_thread_t	alivechecker;

static ECMBOOL stopping_alivechecker;

void
set_MRU_recv_dpch(ebcd_pch_t *dpch)
{
	ecm_uint64	curtick;

	if (ebcd_alivecheck_timeout <= 0 || !dpch->alivecheck)
		return;

	curtick = ecm_tickcount();
	ecm_mutex_lock(alivecheck_mutex);
	dpch->last_recv = curtick;
	if (!list_empty(&dpch->list_recv)) {
		list_del(&dpch->list_recv);
		list_add(&dpch->list_recv, &recv_pchs);
		EBCDDBG(DBG_ALIVECHECK, "%s: MRU recv", ebcd_dbg_pch(dpch));
	}
	ecm_mutex_unlock(alivecheck_mutex);
}

void
set_MRU_sent_dpch(ebcd_pch_t *dpch)
{
	if (ebcd_alivecheck_timeout <= 0 || !dpch->alivecheck)
		return;
	dpch->last_sent = ecm_tickcount();
	ecm_mutex_lock(alivecheck_mutex);
	if (!list_empty(&dpch->list_sent)) {
		list_del(&dpch->list_sent);
		list_add(&dpch->list_sent, &sent_pchs);
	}
	ecm_mutex_unlock(alivecheck_mutex);
}

static ebcd_pch_t *
get_oldest_sent_dpch(void)
{
	ebcd_pch_t	*dpch;
	struct list_head	*lp;

	list_for_each_tail(lp, &sent_pchs) {
		dpch = list_entry(lp, ebcd_pch_t, list_sent);
		if (!dpch->destroyed)
			return dpch;
	}
	return NULL;
}

static ecm_uint32
check_stale_sent_dpch(void)
{
	ebcd_pch_t	*dpch;
	ecm_uint32	dur_nosent;
	ECMBOOL		pending = ECM_FALSE;
	ecm_uint64	curtick;
	ebc_err_t	err;

again:
	curtick = ecm_tickcount();
	dpch = get_oldest_sent_dpch();
	if (dpch == NULL)
		return ebcd_alivecheck_timeout;

	if (curtick < dpch->last_sent) {
		if (pending)
			return 0;
		return ebcd_alivecheck_timeout;
	}
	dur_nosent = (ecm_uint32)(curtick - dpch->last_sent);
	if (dur_nosent >= (ecm_uint32)ebcd_alivecheck_timeout) {
		EBCD_PCH_GET_ASSERT(dpch);
		ecm_mutex_unlock(alivecheck_mutex);
		err = ebcd_send_aliveness(dpch);
		switch (err) {
		case -ERR_EBC_DISCONN:
			EBCDDBG(DBG_ALIVECHECK, "%s: cannot send aliveness packet", ebcd_dbg_pch(dpch));
			ECMERR("%s: cannot check aliveness", dpch->pchp->name);
			ebcd_destroy_pch(dpch);
			ebcd_pch_put(dpch);
			break;
		case EBC_OK:
			ebcd_pch_put(dpch);
			break;
		default:
			ebcd_pch_put(dpch);
			pending = ECM_TRUE;
			break;
		}
		ecm_mutex_lock(alivecheck_mutex);
		goto again;
	}

	if (pending)
		return 0;
	return (ebcd_alivecheck_timeout - dur_nosent);
}

static ebcd_pch_t *
get_oldest_recv_dpch(void)
{
	ebcd_pch_t	*dpch;
	struct list_head	*lp;

	list_for_each_tail(lp, &recv_pchs) {
		dpch = list_entry(lp, ebcd_pch_t, list_recv);
		if (!dpch->destroyed)
			return dpch;
	}
	return NULL;
}

static ecm_uint32
check_stale_recv_dpch(void)
{
	ebcd_pch_t	*dpch;
	ecm_uint64	curtick;
	ecm_uint32	timeout_max_recv, dur_norecv;

	timeout_max_recv = ebcd_alivecheck_timeout * 2;
again:
	dpch = get_oldest_recv_dpch();
	if (dpch == NULL)
		return ebcd_alivecheck_timeout;

	curtick = ecm_tickcount();
	dur_norecv = (ecm_uint32)(curtick - dpch->last_recv);
	if (dur_norecv > timeout_max_recv) {
		EBCD_PCH_GET_ASSERT(dpch);
		ecm_mutex_unlock(alivecheck_mutex);
		EBCDDBG(DBG_ALIVECHECK, "%s: seems to be dead: %umsec", ebcd_dbg_pch(dpch), dur_norecv);
		ECMERR("%s: node no response", dpch->pchp->name);
		ebcd_destroy_pch(dpch);
		ebcd_pch_put(dpch);
		ecm_mutex_lock(alivecheck_mutex);
		goto again;
	}
	return (timeout_max_recv - dur_norecv);
}

#define MIN(a, b)	((a) < (b) ? (a): (b))

static void
check_aliveness(ecm_ptr_t arg)
{
	ecm_uint32	timeout;
	ecm_uint32	timeout_sent, timeout_recv;

	ecm_mutex_lock(alivecheck_mutex);

	while (!stopping_alivechecker) {
		timeout_sent = check_stale_sent_dpch();
		timeout_recv = check_stale_recv_dpch();

		timeout = MIN(timeout_sent, timeout_recv);

		ecm_cond_wait(alivecheck_cond, alivecheck_mutex, &timeout);
	}

	ecm_mutex_unlock(alivecheck_mutex);
}

void
ebcd_init_alivecheck(ebcd_pch_t *dpch)
{
	ecm_uint64	curtick = ecm_tickcount();

	if (ebcd_alivecheck_timeout <= 0)
		return;

	ecm_mutex_lock(alivecheck_mutex);
	dpch->last_sent = curtick;
	dpch->last_recv = curtick;
	list_add(&dpch->list_sent, &sent_pchs);
	list_add(&dpch->list_recv, &recv_pchs);
	ecm_mutex_unlock(alivecheck_mutex);
}

void
ebcd_clear_alivecheck(ebcd_pch_t *dpch)
{
	if (ebcd_alivecheck_timeout <= 0)
		return;

	ecm_mutex_lock(alivecheck_mutex);
	list_del_init(&dpch->list_sent);
	list_del_init(&dpch->list_recv);
	ecm_mutex_unlock(alivecheck_mutex);
}

void
ebcd_start_alivechecker(void)
{
	if (ebcd_alivecheck_timeout <= 0)
		return;

	alivecheck_mutex = ecm_mutex_create(ECM_FALSE);
	alivecheck_cond = ecm_cond_create();
	stopping_alivechecker = ECM_FALSE;
	ecm_thread_create(&alivechecker, check_aliveness, NULL);
}

void
ebcd_stop_alivechecker(void)
{
	if (ebcd_alivecheck_timeout <= 0)
		return;

	ecm_mutex_lock(alivecheck_mutex);
	stopping_alivechecker = ECM_TRUE;
	ecm_cond_signal(alivecheck_cond);
	ecm_mutex_unlock(alivecheck_mutex);

	ecm_thread_free(alivechecker);
	ecm_mutex_free(alivecheck_mutex);
	ecm_cond_free(alivecheck_cond);
}
